package com.chief.hadoop.mr.outputFormat;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.viewfs.ViewFileSystem;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

public class FlowOutPutRecordWriter extends RecordWriter<Text, FlowBean> {

    FSDataOutputStream sj, gh;

    public FlowOutPutRecordWriter(TaskAttemptContext job) {
        try {
            FileSystem fileSystem = ViewFileSystem.newInstance(job.getConfiguration());
            sj = fileSystem.create(new Path("/data1/sj.txt"));
            gh = fileSystem.create(new Path("/data1/gh.txt"));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void write(Text key, FlowBean value) throws IOException, InterruptedException {
        if (key.getLength() > 10) {
            sj.writeBytes(key + "-----" + value.getTotal() + "-----" + value.getCnt() + "\n");
        } else {
            gh.writeBytes(key + "-----" + value.getTotal() + "-----" + value.getCnt() + "\n");
        }
    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        IOUtils.closeStreams(sj, gh);
    }
}
